package org.zjvis.datascience.service;

import cn.hutool.core.util.CharUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
import com.alibaba.druid.util.Base64;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.google.common.base.Joiner;

import java.io.*;
import java.net.HttpCookie;
import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.*;
import java.util.stream.Collectors;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.zjvis.datascience.common.dto.JhubDTO;
import org.zjvis.datascience.common.dto.ProjectDTO;
import org.zjvis.datascience.common.dto.TaskDTO;
import org.zjvis.datascience.common.dto.TaskInstanceDTO;
import org.zjvis.datascience.common.enums.FileTypeEnum;
import org.zjvis.datascience.common.exception.DataScienceException;
import org.zjvis.datascience.common.util.JwtUtil;
import org.zjvis.datascience.common.util.SqlUtil;
import org.zjvis.datascience.common.util.task.TaskUtil;
import org.zjvis.datascience.common.vo.TaskVO;
import org.zjvis.datascience.service.dag.TaskRunnerResult;
import org.zjvis.datascience.service.dataprovider.GPDataProvider;
import org.zjvis.datascience.service.mapper.JhubMapper;
import org.zjvis.datascience.service.mapper.ProjectMapper;
import org.zjvis.datascience.service.mapper.TaskInstanceMapper;
import org.zjvis.datascience.service.mapper.TaskMapper;
import org.zjvis.datascience.service.csv.ColumnType;
import org.zjvis.datascience.service.csv.postgresql.PostgresqlTableCreator;
import org.zjvis.datascience.service.csv.postgresql.PostgresqlCsvTransformer;

import tech.tablesaw.api.Table;
import tech.tablesaw.io.csv.CsvReadOptions;

@Service
public class JhubService {

    private final static Logger logger = LoggerFactory.getLogger("JhubService");

    @Value("${Jhub.host}")
    private String host;

    @Value("${Jhub.blockSize}")
    private int blockSize;

    @Value("${Jhub.adminUser}")
    private String adminUser;

    @Value("${Jhub.adminPassword}")
    private String adminPassword;

    @Value("${postgres.port}")
    private int gpServerPort;


    @Value("${postgres.host}")
    private String gpServerHost;

    @Value("${postgres.username}")
    private String gpServerUserName;

    @Value("${postgres.database}")
    private String gpServerDBName;

    @Value("${postgres.password}")
    private String gpServerPassword;

    @Value("${upload.folder-path}")
    private String filepathRootDir;

    @Autowired
    private TaskMapper taskMapper;

    @Autowired
    private JhubMapper jhubMapper;

    @Autowired
    private ProjectMapper projectMapper;

    @Autowired
    private TaskInstanceMapper taskInstanceMapper;

    @Autowired
    private GPDataProvider gpDataProvider;

    private String adminToken;


    /**
     * 初始化拿adminToken
     *
     */
    public void init() throws Exception {

        String url = host + "/hub/login";
        Map<String, Object> formDataMap = new JSONObject();
        formDataMap.put("username", adminUser);
        formDataMap.put("password", adminPassword);
        logger.info("JHub urlGetCookie: " + url);
        logger.info("JHub request form data: " + formDataMap);
        HttpRequest requestGetCookie = HttpUtil.createPost(url).form(formDataMap);
        logger.info("JHub request get cookie:" + requestGetCookie);
        HttpResponse respGetCookie = requestGetCookie.execute();
        if (respGetCookie.getStatus() != 302 && respGetCookie.getStatus() != 200){
            throw new Exception("failed to get admin cookie");
        }
        List<HttpCookie> cookies = respGetCookie.getCookies();

        String urlGetToken = host + "/hub/api/users/admin/tokens";
        String bodyStr = "{\"note\":\"Requested via token page\",\"expires_in\":null}";
        logger.info("JHub urlGetToken: " + urlGetToken);
        logger.info("JHub bodyStr: " + bodyStr);
        logger.info("JHub cookies: " + cookies);
        //HttpCookie mergedCookie = cookies.
        HttpRequest request2GetToken = HttpUtil.createPost(urlGetToken).body(bodyStr).cookie(cookies.toArray(new HttpCookie[0]))
                .header("referer",host + "/hub/token");
        logger.info("JHub httpRequest: " + request2GetToken);
        HttpResponse resp2GetToken = request2GetToken.execute();
        logger.info("JHub httpResponse: " + resp2GetToken);
        if (resp2GetToken.getStatus() != 200){
            throw new Exception("failed to get admin token");
        }
        adminToken = JSONObject.parseObject(resp2GetToken.body()).getString("token");
        logger.info("JHub admin token:" + adminToken);

    }

    /**
     * 根据userId获取jupyterHub信息
     *
     * @param userId
     * @return
     */
    public JhubDTO queryByUserId(Long userId) {
        return jhubMapper.queryByUserId(userId);
    }

    /**
     * 获取juypterHub url
     *
     * @param userId
     * @param taskId
     * @return
     */
    public String getJupyterHubWorkspaceUrl(Long userId, Long taskId) {
        checkUserId(userId);

        String token = getJupyterHubUserToken(userId);

        String path = generatePath(taskId);
        String name = generateJupyterHubUserName(userId);
        //https://jupyterhub-ops-test.lab.zjvis.net:32443/user/admin/lab/workspaces/auto-O/tree/fff?token=b4b9dcd02

        return "/user/" + name + "/lab/workspaces/auto-O/tree/" + path + "?token=" + token;
    }

    /**
     * 获取jupyterHub用户api token，不存在则生成token并存入mysql
     *
     * @param userId
     * @return
     */
    public String getJupyterHubUserToken(Long userId) {
        checkUserId(userId);
        //服务不在则启动
        if (!serverIsReady("user_" + userId, 1)){
            //start
            startServer(userId);
        }

        JhubDTO jhub = jhubMapper.queryByUserId(userId);

        if (jhub == null) {
            String username = generateJupyterHubUserName(userId);
            String url = host + "/hub/api/users/" + username + "/tokens";
            HttpResponse resp = HttpUtil.createPost(url)
                    .header("Authorization", "token " + adminToken)
                    .execute();

            if (resp.getStatus() == 200) {
                jhub = new JhubDTO();
                jhub.setUserId(userId);
                jhub.setToken(JSONObject.parseObject(resp.body()).getString("token"));
                logger.info("user " + userId + " got api token from jupyterHub" + jhub.getToken());
                jhubMapper.insert(jhub);
            } else {
                logger.error("status: " + resp.getStatus() + " body: " + resp.body());
                throw new DataScienceException(
                        JSONObject.parseObject(resp.body()).getString("message"));
            }
        }
        return jhub.getToken();
    }

    /**
     * 判断用户id
     *
     * @param userId
     */
    public void checkUserId(Long userId) {
        if (userId == null || userId == 0) {
            throw new DataScienceException("userId不存在");
        }
    }

    /**
     * 创建用户在jupyterHub中的path
     *
     * @param task
     * @param username
     * @param token
     * @return
     */
    public String createJupyterHubPath(TaskDTO task, String username, String token) {
        List<String> paths = generatePathList(task);
        return createJupyterHubPath(paths, username, token);
    }

    /**
     * 创建用户在jupyterHub中的path
     *
     * @param paths    层级path（单次api调用只能创建一个目录）
     * @param username
     * @param token
     * @return
     */
    public String createJupyterHubPath(List<String> paths, String username, String token) {
        if (StringUtils.isBlank(username)) {
            throw new DataScienceException("username不能为空");
        }
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        String path = "";

        for (int i = 0; i < paths.size(); i++) {
            String url;
            path = paths.get(i);
            if (pathExist(username, paths.get(i), token)) {
                continue;
            }
            JSONObject bd = new JSONObject();
            if (i == 0) {
                bd.put("path", "");
                url = host + "/user/" + username + "/api/contents";
            } else {
                bd.put("path", paths.get(i - 1));
                url = host + "/user/" + username + "/api/contents/" + paths.get(i - 1);
            }

            logger.info("create path url: " + url);

            bd.put("type", "directory");

            HttpResponse resp = HttpUtil.createPost(url)
                    .header("Authorization", "token " + token)
                    .body(bd.toJSONString())
                    .execute();
            System.err.println("create status: " + resp.getStatus());
            System.err.println(resp.body());

            if (resp.getStatus() == 201) {
                JSONObject res = JSONObject.parseObject(resp.body());
                String oldName = res.getString("path");
                String newName = paths.get(i);
                renameDir(username, token, oldName, newName);
            } else {
                logger.warn("status: " + resp.getStatus() + " body: " + resp.body());
            }
        }
        return legalize(path);
    }

    /**
     * 重命名task文件夹
     *
     * @param vo
     * @param oldDirName
     */
    public void renameTaskDir(TaskVO vo, String oldDirName) {
        String projectName = projectMapper.selectByPrimaryKey(vo.getProjectId()).getName() + "/";
        String newDirName = vo.getName() + "_" + vo.getId();
        Long userId = JwtUtil.getCurrentUserId();
        String token = getJupyterHubUserToken(userId);
        String username = "user_" + userId;
        renameDir(username, token, legalize(projectName + oldDirName + "_" + vo.getId()), legalize(projectName + newDirName));
    }

    /**
     * 重命名project文件夹
     *
     * @param projectDTO
     * @param newDirName
     */
    public void renameProjectDir(ProjectDTO projectDTO, String newDirName) {
        String oldDirName = projectDTO.getName();
        Long userId = JwtUtil.getCurrentUserId();
        String token = getJupyterHubUserToken(userId);
        String username = "user_" + userId;
        renameDir(username, token, legalize(oldDirName), legalize(newDirName));
    }

    /**
     * 重命名目录
     *
     * @param username
     * @param token
     * @param oldName
     * @param newName
     */
    public void renameDir(String username, String token, String oldName, String newName) {
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        JSONObject body = new JSONObject();

        String url = host + "/user/" + username + "/api/contents/" + oldName;
        body.put("path", legalize(newName));
        logger.info("jupyterHub rename: " + url);

        HttpResponse resp = HttpUtil.createRequest(Method.PATCH, url)
                .header("Authorization", "token " + token)
                .body(body.toJSONString())
                .execute();

        logger.info("rename status: " + resp.getStatus() + " body: " + resp.body());
    }

    /**
     * 下划线替换名字中给括号
     *
     * @param dirName
     * @return
     */
    private String legalize(String dirName) {
        return dirName.replace(")", "")
                .replace("(", "_")
                .replace(" ", "_");
    }

    /**
     * 生成完整目录
     *
     * @param taskId
     * @return
     */
    private String generatePath(Long taskId) {
        TaskDTO task = taskMapper.queryById(taskId);
        if (task == null) {
            throw new DataScienceException("task不存在");
        }

        return generatePath(task.getProjectId(), taskId);
    }

    /**
     * 生成完整目录
     *
     * @param projectId
     * @param taskId
     * @return
     */
    private String generatePath(Long projectId, Long taskId) {
        String projectName = projectMapper.getById(projectId).getName();
        String taskName = taskMapper.queryById(taskId).getName();
        return legalize(projectName) + "/" + legalize(taskName) + "_" + taskId;
    }

    /**
     * 从父目录开始，生成每层path列表
     *
     * @param taskId
     * @return
     */
    private List<String> generatePathList(Long taskId) {
        TaskDTO task = taskMapper.queryById(taskId);
        return generatePathList(task);
    }

    /**
     * 从父目录开始，生成每层path列表
     *
     * @param task
     * @return
     */
    private List<String> generatePathList(TaskDTO task) {
        if (task == null) {
            throw new DataScienceException("task不存在");
        }
        ProjectDTO project = projectMapper.selectByPrimaryKey(task.getProjectId());
        if (project == null) {
            throw new DataScienceException("project不存在");
        }
        List<String> paths = new ArrayList<>();
        String projectDirName = project.getName();
        paths.add(projectDirName);

        String folderName = projectDirName + "/" + task.getName() + "_" + task.getId();
        paths.add(legalize(folderName));
        return paths;
    }

    public List<String> generatePathList(String urlPath) {
        if (StringUtils.isBlank(urlPath)) {
            throw new DataScienceException("urlPath不能为空");
        }
        List<String> paths = new ArrayList<>();
        String[] urlPathParts = urlPath.split("\\/");
        String tmpPath = "";
        if (urlPathParts.length > 1) {
            for (int i = 0; i < urlPathParts.length; i++) {
                if (i == 0) {
                    tmpPath += urlPathParts[i];
                } else {
                    tmpPath += "/" + urlPathParts[i];
                }
                paths.add(tmpPath);
            }
        } else {
            paths.add(urlPath);
        }
        return paths;
    }

    /**
     * 判断目录是否存在
     *
     * @param username
     * @param path
     * @param token
     * @return
     */
    public boolean pathExist(String username, String path, String token) {
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        String url = host + "/user/" + username + "/api/contents/" + path;
        logger.info("check path url:" + url);
        logger.info("check path url token:" + token);

        HttpResponse resp = HttpUtil.createGet(url)
                .header("Authorization", "token " + token)
                .execute();

        if (resp.getStatus() == 200) {
            logger.info(path + " exist");
            return true;
        } else if (resp.getStatus() == 404) {
            logger.info(path + " not exist");
            return false;
        } else {
            logger.error("status:" + resp.getStatus() + ", body:" + resp.body());
            throw new DataScienceException("status:" + resp.getStatus() + ", body:" + resp.body());
        }
    }

    /**
     * 从jupyterHub获取文件内容
     *
     * @param token    用户api token
     * @param username 用户名
     * @param path     文件在jupyterHub上的路径
     * @return jupyterHub上的文件内容
     */
    public String getFileContentFromJupyterHub(String token, String username, String path) {
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        String url = host + "/user/" + username + "/files/" + path;
        logger.info("download url: " + url);

        HttpResponse resp = HttpUtil.createGet(url)
                .header("Authorization", "token " + token)
                .execute();

        if (resp.getStatus() == 200) {
            return resp.body();
        } else if (resp.getStatus() == 404) {
            logger.info(path + " not exist");
            throw new DataScienceException("要下载的文件不存在：" + path);
        } else {
            logger.error("status:" + resp.getStatus() + ", body:" + resp.body());
            throw new DataScienceException("status:" + resp.getStatus() + ", body:" + resp.body());
        }
    }

    /**
     * 从gp下载数据
     *
     * @param schema gp数据库下的schema
     * @param table  数据库表名
     * @return
     */
    public String downloadFromGP(String schema, String table) {
        if (table.contains(".")) {
            String[] tmps = table.split("[.]");
            table = tmps[1];
            schema = tmps[0];
        }

        Map<String, String> meta = gpDataProvider.getTableMetaMap(table.replaceAll("\"", ""), schema);
        List<String> cols = meta.keySet().stream().filter(name -> !"_record_id_".equals(name))
                .collect(Collectors.toList());
//        List<String> cols = new Arr
//        cols.addAll(meta.keySet());ayList<>();

        JSONArray jsonArray = gpDataProvider.executeQuery("select * from " + schema + "."+ table.replaceAll("\"", "") + " order by _record_id_");
        InputStream in = new ByteArrayInputStream(jsonArray.toJSONString().getBytes(StandardCharsets.UTF_8));

        JSONArray data;
        try {
            data = JSON.parseObject(in, JSONArray.class, Feature.OrderedField);
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
            throw new DataScienceException("从gp下载文件失败");
        }
        StringBuffer sb = new StringBuffer();
        sb.append(Joiner.on(",").join(SqlUtil.formatPGSqlCols(cols))).append("\n");
        for (int i = 0; i < data.size(); ++i) {
            JSONObject item = data.getJSONObject(i);
            List<String> values = new ArrayList<>();
            for (String col : cols) {
                values.add(item.getString(col) == null ? ""
                        : SqlUtil.formatPGSqlColName(item.getString(col)));
            }
            sb.append(Joiner.on(",").join(values)).append("\n");
        }
        return sb.toString();
    }

    /**
     * 上传文件到jupyterHub
     *
     * @param token    用户api token
     * @param username 用户名
     * @param path     带文件名的路径
     * @param fileName 文件名
     * @param content  文件内容
     * @param chunk    第n块
     */
    public void uploadFileToJupyterHub(String token, String username, String path, String fileName,
                                       String content, Integer chunk) {
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        //上传内容转base64格式
        content = Base64.byteArrayToBase64(content.getBytes());

        logger.info("chunk:" + chunk);

        String url =
                host + "/user/" + username + "/api/contents/" + path + "?" + System.currentTimeMillis();
        System.err.println("url: " + url);

        //生成请求体
        JSONObject body = new JSONObject();
        body.put("type", "file");
        body.put("format", "base64");
        body.put("name", fileName);
        body.put("content", content);
        body.put("path", path);
        if (chunk != null) {
            body.put("chunk", chunk);
        }

        deleteJupyterHubContent(token, username, path);

        HttpResponse resp = HttpUtil.createRequest(Method.PUT, url)
                .header("Authorization", "token " + token)
                .body(body.toJSONString())
                .execute();

        logger.info("upload status: " + resp.getStatus() + " body: " + resp.body());
    }


    /**
     * 获取gp数据并写入jupyter
     *
     * @param parenTask
     * @param username
     * @param token
     * @param path
     */
    public void gpDataToJupyterHub(TaskDTO parenTask, String username, String token, String path) {
        String schema = "dataset";
        String table = TaskUtil.extractTableStr(parenTask);
        String content = downloadFromGP(schema, table);
        if (StringUtils.isBlank(content)) {
            throw new DataScienceException("文件内容为空");
        }

        //String fileName1 = splitGpSchemaAndTableName(table).get("tableName") + ".csv";
        String fileName = legalize(parenTask.getName() + ".csv");

        path = path + "/" + fileName;

        uploadFileContentToJupyterHub(content, username, token, path, fileName);
    }

    public void uploadFileContentToJupyterHub(String content, String username, String token,
                                              String pathWithFileName, String fileName) {

        int chunk = 1;
        for (int start = 0; start < content.length(); ) {
            int end = (start + blockSize) > content.length() ? content.length() : start + blockSize;
            String chunkContent = content.substring(start, end);

            uploadFileToJupyterHub(token, username, pathWithFileName, fileName, chunkContent,
                    end == content.length() ? -1 : chunk++);
            start += blockSize;
        }
    }

    /**
     * 判断用户服务状态
     *
     * @param userId
     * @return
     */
    public boolean checkUserServer(Long userId) {
        return checkUserServer(generateJupyterHubUserName(userId));
    }

    /**
     * 生成jupyterHub用户名
     *
     * @param userId
     * @return
     */
    public String generateJupyterHubUserName(Long userId) {
        checkUserId(userId);
        return "user_" + userId;
    }

    /**
     * 判断用户服务状态
     *
     * @param username
     * @return
     */
    public boolean checkUserServer(String username) {
        //获取用户信息并转为json格式
        JSONObject jo = JSONObject.parseObject(getUser(username));
        JSONObject res = jo.getJSONObject("servers");
        if (!res.isEmpty() && res.getJSONObject("").getBoolean("ready")) {
            logger.info("server is start ");
            return true;
        } else {
            logger.info("server not start ");
            return false;
        }
    }

    /**
     * 判断服务是否启动好
     *
     * @param username
     * @param retry    重试次数（最低60次）
     * @return
     */
    public boolean serverIsReady(String username, int retry) {
        if (checkUserServer(username)) {
            return true;
        } else {
            if (retry > 0) {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(), e);
                }
                retry -= 1;
                return serverIsReady(username, retry);
            } else {
                return false;
            }
        }
    }

    /**
     * 开启服务
     *
     * @param userId
     */
    public void startServer(Long userId) {
        startServer(generateJupyterHubUserName(userId));
    }

    /**
     * 开启服务
     *
     * @param username
     */
    public void startServer(String username)  {
        if (checkUserServer(username)) {
            return;
        }
        String url = host + "/hub/api/users/" + username + "/server";
        System.err.println("url: " + url);

        HttpResponse resp = HttpUtil.createPost(url)
                .header("Authorization", "token " + adminToken)
                .execute();
        if (resp.getStatus() == 202 && serverIsReady(username, 60)) {
            logger.info("启动服务成功");
        } else {
            logger.error("start server status: " + resp.getStatus() + " body: " + resp.body());
            throw new DataScienceException("启动服务失败");
        }
    }

    public void init(Long userId) {

        //根据项目及task创建，jupyter对应目录
        String username = generateJupyterHubUserName(userId);

        //判断jupyter用户是否存在，不存在则创建一个
        createUser(username);

        //获取用户api token
        getJupyterHubUserToken(userId);

        //开启服务
        startServer(username);

    }

    /**
     * 初始化
     *
     * @param task
     */
    public void init(TaskDTO task) {
        if (task == null) {
            throw new DataScienceException("task节点不存在");
        }

        //根据项目及task创建，jupyter对应目录
        String username = generateJupyterHubUserName(task.getUserId());

        //判断jupyter用户是否存在，不存在则创建一个
        createUser(username);

        //获取用户api token
        String token = getJupyterHubUserToken(task.getUserId());

        //开启服务
        startServer(username);

        createJupyterHubPath(task, username, token);
    }

    public void loadData(Long userId, TaskDTO parenTask, TaskDTO task) {
        if (task == null) {
            throw new DataScienceException("task节点不存在");
        }

        //根据项目及task创建，jupyter对应目录
        String username = generateJupyterHubUserName(userId);

        //判断jupyter用户是否存在，不存在则创建一个
        createUser(username);

        //获取用户api token
        String token = getJupyterHubUserToken(userId);

        //开启服务
        startServer(username);

        String path = createJupyterHubPath(task, username, token);

        //根据父类信息将gp表数据转csv并同步到jupyterHub
        gpDataToJupyterHub(parenTask, username, token, path);
    }

    /**
     * 获取task在jupyterHub目录下的所有csv文件
     *
     * @param userId
     * @param taskId
     * @return
     */
    public String loadJhubFiles(Long userId, Long taskId) {

        TaskDTO task = taskMapper.queryById(taskId);

        String user = generateJupyterHubUserName(userId);
        String token = getJupyterHubUserToken(userId);

        String url = host + "/user/" + user + "/api/contents/" + generatePath(taskId);

        HttpResponse resp = HttpUtil.createGet(url).header("Authorization", "token " + token)
                .execute();

        logger.info("list file status: " + resp.getStatus() + " body: " + resp.body());

        JSONArray fileArray = new JSONArray();
        List<String> fileNames = new ArrayList<>();

        if (StringUtils.isNotBlank(resp.body())) {
            JSONObject respBodyJson = JSONObject.parseObject(resp.body());
            if (respBodyJson.containsKey("content")) {
                JSONArray contents = respBodyJson.getJSONArray("content");
                if (contents != null && !contents.isEmpty()) {
                    for (int i = 0; i < contents.size(); i++) {
                        String name = contents.getJSONObject(i).getString("name");
                        if (name.endsWith(".csv")) {
                            JSONObject optionPair = new JSONObject();
                            optionPair.put("label", name);
                            optionPair.put("value", name);
                            fileArray.add(optionPair);
                            fileNames.add(name);
                        }
                    }
                }
            }
        }

        //load setParams
        JSONObject data_json = JSONObject.parseObject(task.getDataJson());
        JSONArray setParamsJsonArray = data_json.getJSONArray("setParams");
        JSONObject setParams = setParamsJsonArray.getJSONObject(0);

        setParams.getJSONArray("formItem").getJSONObject(0)
                .getJSONObject("props").put("options", fileArray);

        JSONObject formData = setParams.getJSONObject("formData");
        String filenameFormData = formData.getString("filename");
        if (filenameFormData != null) {
            if (filenameFormData.equals("<暂无可选csv文件>")) {
                if (fileArray.size() != 0) {
                    formData.put("filename", fileNames.get(fileNames.size() - 1));
                } else {
                    formData.put("filename", "<请确保当前节点目录下有csv文件>");
                }
            } else {
                if (!fileNames.contains(filenameFormData)) {
                    formData.put("filename", "<请选择csv文件>");
                }
            }
        }

        task.setDataJson(data_json.toString());
        taskMapper.update(task);

        return fileArray.toJSONString();
    }

    /**
     * 执行数据从jupyterHub到gp数据库任务
     *
     * @param instance
     * @return
     */
    public TaskRunnerResult execRunner(TaskInstanceDTO instance) {
        TaskRunnerResult taskRunnerResult = new TaskRunnerResult();

        String res = jupyterHubFileToGPDB(instance.getTaskId(), instance.getId());

        TaskInstanceDTO instanceNew = taskInstanceMapper.queryById(instance.getId());
        JSONObject logInfo = new JSONObject();
        if ("finished".equals(res)) {
            instance.setDataJson(instanceNew.getDataJson());
            taskRunnerResult.setStatus(0);
            logInfo.put("result", "SUCCESS");
        } else {
            taskRunnerResult.setStatus(500);
            logInfo.put("result", res);
        }

        instance.setLogInfo(logInfo.toString());

        return taskRunnerResult;
    }

    /**
     * jupyterHub文件（csv文件）数据导入到gp数据库
     *
     * @param taskId
     * @param instanceId
     * @return
     */
    public String jupyterHubFileToGPDB(Long taskId, Long instanceId) {
        TaskDTO task = taskMapper.queryById(taskId);
        String filePath = jupyterHubFileToGPServer(task);
        if (StringUtils.isBlank(filePath)) {
            return "finished";
        }
        String tableName = "pipeline.jhub_" + System.currentTimeMillis();
        try {
            csvToPostgres( filePath, tableName, instanceId);
        } catch (SQLException throwables) {
            throwables.printStackTrace();
            logger.error("csv2pg error:" + throwables);
            return String.valueOf(throwables);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error("csv2pg error:" + e);
            return String.valueOf(e);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            logger.error("csv2pg error:" + e);
            return String.valueOf(e);
        }
        //String extra = filePath + " " + tableName + " " + instanceId;
        //return execDFToDB(extra);

        return "finished";
    }

    public void csvToPostgres( String filePath, String targetTableName, Long instanceId) throws SQLException, IOException, ClassNotFoundException {
        File csvFile = new File(filePath);

        Table sawTable = Table.read().csv(CsvReadOptions.builder(csvFile).ignoreZeroDecimal(false).build());
        boolean addIndex = !sawTable.columnNames().contains("_record_id_");

        PostgresqlTableCreator tableCreator = PostgresqlTableCreator.builder()
                .tableName(targetTableName)
                .columnNameList(sawTable.columnNames())
                .columnTypeList(Arrays.stream(sawTable.columnTypes())
                        .map(tech.tablesaw.api.ColumnType::name)
                        .map(ColumnType::valueOf)
                        .collect(Collectors.toList()))
                .addRecordIdIndex(addIndex)
                .build();

        PostgresqlCsvTransformer postgresqlCsvTransformer = PostgresqlCsvTransformer.builder()
                .host(gpServerHost)
                .port(gpServerPort)
                .username(gpServerUserName)
                .password(gpServerPassword)
                .csvFile(csvFile)
                .databaseName(gpServerDBName)
                .tableCreator(tableCreator)
                .columnNameList(addIndex ? sawTable.columnNames() : Collections.emptyList())
                .build();
        postgresqlCsvTransformer.csvToTable();
        //todo update instance table
        TaskInstanceDTO taskInstanceDTO = taskInstanceMapper.queryById(instanceId);
        JSONArray outputArray = new JSONArray();
        JSONObject output = new JSONObject();
        output.put("tableCols",tableCreator.columnNameListOfTable());
        output.put("columnTypes", tableCreator.columnTypeListOfTable());
        output.put("tableName", targetTableName);
        JSONObject semantics = new JSONObject();
        for (String col:tableCreator.columnNameListOfTable()){
            semantics.put(col, null);
        }
        output.put("semantic", semantics);
        output.put("totalRow", sawTable.rowCount());
//        taskInstanceDTO.getDataJson().
        JSONObject dataJson = JSONObject.parseObject(taskInstanceDTO.getDataJson());
        JSONObject inputInfo = dataJson.getJSONObject("inputInfo");
        outputArray.add(output);
        inputInfo.put("output",outputArray);
        dataJson.put("inputInfo",inputInfo);
        taskInstanceDTO.setDataJson(dataJson.toJSONString());
        taskInstanceDTO.setOutput(output.toJSONString());
        taskInstanceMapper.update(taskInstanceDTO);
        //JSONObject.parseObject(taskInstanceDTO.getDataJson()).put("output")
        //delete local file
        FileUtils.deleteQuietly(csvFile);
    }

    /**
     * jupyterHub文件下载到gp服务器
     *
     * @param task task信息
     * @return 下载到gp服务器的文件路径
     */
    public String jupyterHubFileToGPServer(TaskDTO task) {
        //先判断文件名是否存在
        String jupyterFileName = getFileName(task);
        if (StringUtils.isBlank(jupyterFileName) || "<暂无可选csv文件>".equals(jupyterFileName)
                || "<请确保当前节点目录下有csv文件>".equals(jupyterFileName)) {
            return null;
        }

        Long userId = task.getUserId();
        String fileName = "user_" + userId + "_" + task.getId() + "_" + System.currentTimeMillis();
        String token = getJupyterHubUserToken(userId);
        String username = generateJupyterHubUserName(userId);

        //生成jupyterHub文件路径
        String jupyterHubFilePath =
                projectMapper.getById(task.getProjectId()).getName() + "/" + task.getName() + "_" + task.getId() + "/" + jupyterFileName;
        jupyterHubFilePath = legalize(jupyterHubFilePath);

        String content = getFileContentFromJupyterHub(token, username, jupyterHubFilePath);
//gpServerFileInGPServerHomeDir
        File dir = new File(filepathRootDir);
        if (!dir.exists()) {
            dir.mkdirs();
        }
        String filePath = filepathRootDir + "/jhub_" + fileName + CharUtil.DOT + FileTypeEnum.CSV.getValue();
        InputStream in = new ByteArrayInputStream(content.getBytes());

        //todo 下载content到本地csv
        File targetfile = new File(filePath);
        try {
            logger.info("saving file to path" + filePath);
            copyInputStreamToFile(in, targetfile);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error("saving content error: " + e);
        }
        //上传文件到指定目录
        //sftpConnectService.uploadFile(in, filePath);
        return filePath;
    }


    private static void copyInputStreamToFile(InputStream inputStream, File file)
            throws IOException {

        // append = false
        try (FileOutputStream outputStream = new FileOutputStream(file, false)) {
            int read;
            byte[] bytes = new byte[8192];
            while ((read = inputStream.read(bytes)) != -1) {
                outputStream.write(bytes, 0, read);
            }
        }

    }


    /**
     * 从task中获取文件名
     *
     * @param task
     * @return
     */
    private String getFileName(TaskDTO task) {
        JSONObject data_json = JSONObject.parseObject(task.getDataJson());
        JSONArray setParamsJsonArray = data_json.getJSONArray("setParams");
        JSONObject setParams = setParamsJsonArray.getJSONObject(0);
        JSONObject formData = setParams.getJSONObject("formData");
        String filenameFormData = formData.getString("filename");
        return legalize(filenameFormData);
    }

    /**
     * 旧task节点对应的数据文件复制到新的task
     *
     * @param sourceTask
     * @param targetTask
     * @param includeFile
     */
    public void copyPaste(TaskDTO sourceTask, TaskDTO targetTask, boolean includeFile) {
        String username = generateJupyterHubUserName(sourceTask.getUserId());
        String sourceUserToken = getJupyterHubUserToken(sourceTask.getUserId());
        String targetUserToken = "";
        if (sourceTask.getUserId().equals(targetTask.getUserId())) {
            targetUserToken = sourceUserToken;
            init(sourceTask.getUserId());
        } else {
            targetUserToken = getJupyterHubUserToken(targetTask.getUserId());
            init(targetTask.getUserId());
        }

        String sourcePath = generatePath(sourceTask.getProjectId(), sourceTask.getId());

        String newPath = generatePath(targetTask.getProjectId(), targetTask.getId());
        List<String> paths = generatePathList(newPath);

        createJupyterHubPath(paths, username, targetUserToken);

        if (includeFile) {
            List<Map<String, String>> list = listJupyterHubFile(username, sourcePath, sourceUserToken);
            if (list != null && !list.isEmpty()) {
                for (Map<String, String> file : list) {
                    String content = getFileContentFromJupyterHub(sourceUserToken, username, file.get("path"));
                    String pathWithFileName = newPath + "/" + file.get("name");
                    uploadFileContentToJupyterHub(content, username, targetUserToken, pathWithFileName, file.get("name"));
                }
            }
        }
    }

    /**
     * 获取jupyterHub文件列表
     *
     * @param username
     * @param path
     * @param token
     * @return
     */
    public List<Map<String, String>> listJupyterHubFile(String username, String path,
                                                        String token) {
        if (StringUtils.isBlank(token)) {
            throw new DataScienceException("token不能为空");
        }

        String url = host + "/user/" + username + "/api/contents/" + path;

        HttpResponse resp = HttpUtil.createGet(url)
                .header("Authorization", "token " + token)
                .execute();

        List<Map<String, String>> list = new ArrayList<>();
        if (resp.getStatus() == 200) {
            String body = resp.body();
            JSONObject bj = JSONObject.parseObject(body);
            JSONArray content = bj.getJSONArray("content");
            if (content != null || !content.isEmpty()) {
                for (int i = 0; i < content.size(); i++) {
                    JSONObject res = content.getJSONObject(i);
                    Map<String, String> map = new HashMap<>();
                    if (!"directory".equals(res.getString("type"))) {
                        map.put("name", res.getString("name"));
                        map.put("path", res.getString("path"));
                        list.add(map);
                    }
                }
            }

            logger.info(path + " exist");
            return list;
        } else if (resp.getStatus() == 404) {
            logger.info(path + " not exist");
            return list;
        } else {
            logger.error("status:" + resp.getStatus() + ", body:" + resp.body());
            throw new DataScienceException("status:" + resp.getStatus() + ", body:" + resp.body());
        }
    }

    /**
     * 删除jupyterHub文件或目录
     *
     * @param token
     * @param username
     * @param path
     */
    public void deleteJupyterHubContent(String token, String username, String path) {
        String url =
                host + "/user/" + username + "/api/contents/" + path;
        System.err.println("url: " + url);

        HttpResponse resp = HttpUtil.createRequest(Method.DELETE, url)
                .header("Authorization", "token " + token)
                .execute();
        if (resp.getStatus() == 204) {
            logger.info("删除成功：" + path);
        } else {
            logger.info("删除失败：" + path);
            logger.warn("status: " + resp.getStatus() + " body: " + resp.body());
        }
    }


    /**
     * 获取用户信息
     *
     * @param name
     * @return
     */
    public String getUser(String name) {
        if (StringUtils.isBlank(name)) {
            return "用户名不能为空";
        }
        String url = host + "/hub/api/users/" + name;
        HttpResponse resp = HttpUtil.createGet(url)
                .header("Authorization", "token " + adminToken)
                .execute();
        if (resp.getStatus() == 409) {
            return JSONObject.parseObject(resp.body()).getString("message");
        }

        logger.info("get user info status: " + resp.getStatus() + " body: " + resp.body());

        return resp.body();
    }

    /**
     * 用户是否存在
     *
     * @param name
     * @return
     */
    public boolean userExist(String name) {
        if (StringUtils.isBlank(name)) {
            throw new DataScienceException("用户名不能为空");
        }
        String url = host + "/hub/api/users/" + name;
        HttpResponse resp = HttpUtil.createGet(url)
                .header("Authorization", "token " + adminToken)
                .execute();
        System.err.println("status: " + resp.getStatus());
        if (resp.getStatus() == 200) {
            return true;
        } else if (resp.getStatus() == 404) {
            return false;
        } else {
            logger.error("status: " + resp.getStatus() + " body: " + resp.body());
            throw new DataScienceException("status: " + resp.getStatus() + " body: " + resp.body());
        }
    }

    /**
     * 创建用户
     *
     * @param name
     * @return
     */
    public String createUser(String name) {
        if (StringUtils.isBlank(name)) {
            return "用户名不能为空";
        }
        if (userExist(name)) {
            return "用户已存在";
        }
        String url = host + "/hub/api/users/" + name;
        HttpResponse resp = HttpUtil.createPost(url)
                .header("Authorization", "token " + adminToken)
                .execute();
        System.err.println("url: " + url);
        System.err.println("status: " + resp.getStatus());
        if (resp.getStatus() == 409) {
            return JSONObject.parseObject(resp.body()).getString("message");
        }
        System.err.println(resp.body());

        return resp.body();
    }

}
